package com.xhs;

import com.xhs.bolt.LogParseBolt;
import com.xhs.bolt.ProductCountBolt;
import com.xhs.spout.AccessLogSendKafkaSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * Created by jm on 2017/11/20.
 */
public class HotProductTopology {


    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("AccessLogSendKafkaSpout",new AccessLogSendKafkaSpout(),1);

        builder.setBolt("LogParseBolt",new LogParseBolt(),1).setNumTasks(1).shuffleGrouping("AccessLogSendKafkaSpout");

        builder.setBolt("ProductCountBolt",new ProductCountBolt(),1).setNumTasks(1).fieldsGrouping("LogParseBolt",new Fields("productId"));

        Config config = new Config();

        if (args!=null && args.length > 0){
            config.setNumWorkers(2);

            try {
                StormSubmitter.submitTopology(args[0],config,builder.createTopology());
            } catch (Exception e) {
                e.printStackTrace();
            }

        }else {
            config.setMaxTaskParallelism(5);

            LocalCluster cluster = new LocalCluster();

            cluster.submitTopology("HotProductCount",config,builder.createTopology());

            try {
                Thread.sleep(3600 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            cluster.shutdown();
        }


    }



}
